from kafka import KafkaConsumer

if __name__ == '__main__':

    #  earliest
    #  当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
    #  latest  默认
    # 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产认值生的该分区下的数据
    #  none
    #  topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常

    # 1、创建消费者
    consumer = KafkaConsumer(
        bootstrap_servers="master:9092,node1:9092,node2:9092",
        group_id="asd22ss2asd",  # 同一条数据子啊一个组内著读取一次
        auto_offset_reset="earliest"
    )
    # 订阅topic
    consumer.subscribe(topics=["student"])
    # 从kafka拉取数据，1秒超时

    while True:
        print("读取数据")
        kv_records = consumer.poll(timeout_ms=1000)

        # 循环解析数据
        for t, records in kv_records.items():
            for record in records:
                # topic
                topic = record.topic
                # 偏移量
                offset = record.offset
                # 分区编号
                partition = record.partition
                # 数据需要解码
                value = record.value.decode("UTF-8")
                # 截取数据
                print(topic, offset, partition, value)
